package org.alitouka.spark.dbscan

import org.alitouka.spark.dbscan.spatial.Point
import org.alitouka.spark.dbscan.util.io.IOHelper
import org.apache.spark.SparkContext
import org.omg.IOP.IORHelper

object DbscanDemo {

  def create2DPoint (x: Double, y: Double, idx: PointId = 0): Point = {
    new Point ( new PointCoordinates (Array (x, y)), idx, 1, Math.sqrt (x*x+y*y))
  }

  def createMultiDPoint(dim: Array[Double], idx: PointId = 0): Point = {
    var sum = 0.00
    for (a <- dim) {
      sum += a*a
    }
    new Point(new PointCoordinates(dim), idx, 1, Math.sqrt(sum))
  }

  def main(args: Array[String]): Unit = {
    val sc = new SparkContext("local[2]", "dbscanDemo")
    val path = "/Users/wangyizhong/t.csv"
    val data = IOHelper.readDataset(sc, path)

    val clusteredPoints = sc.parallelize (Array ( create2DPoint(0.0, 0.0).withClusterId(1),
      create2DPoint(1.0, 0.0).withClusterId(1),
      create2DPoint(0.0, 1.0).withClusterId(1),
      create2DPoint(1.0, 1.0).withClusterId(1),

      create2DPoint(3.0, 0.0).withClusterId(2),
      create2DPoint(4.0, 0.0).withClusterId(2),
      create2DPoint(3.0, 1.0).withClusterId(2),
      create2DPoint(4.0, 1.0).withClusterId(2),

      create2DPoint(1.0, 3.0).withClusterId(DbscanModel.NoisePoint),
      create2DPoint(3.0, 3.0).withClusterId(DbscanModel.NoisePoint)
    ))

    var rdd = sc.parallelize(Array(
      createMultiDPoint(Array(1,2,3,4)),
      createMultiDPoint(Array(2,3,4,5)),
      createMultiDPoint(Array(1,3,2,4)),
      createMultiDPoint(Array(1,4,3,4))
    ))


    val settings1 = new DbscanSettings ().withEpsilon(1.5).withNumberOfPoints(3)
//    val model1 = new DbscanModel (clusteredPoints, settings1)
    val model = Dbscan.train(data, settings1)

    var predictLable = model.allPoints.collect()
    for (a <- predictLable) {
      System.out.println(a.toString())
    }

//    val settings2 = new DbscanSettings ().withEpsilon(1.5).withNumberOfPoints(4).withTreatBorderPointsAsNoise(true)
//    val model2 = new DbscanModel (clusteredPoints, settings2)


//
//    val predictRdd = rdd.map((x:Point) => model.predict(x)).collect()
//    for (a <- predictRdd) {
//      System.out.println(a.toString)
//    }

//    val predictedClusterId = model.predict(createMultiDPoint(Array(1,2,3,4)))
//    System.out.println(predictedClusterId.toString)
  }
}
